package main

import (
	"encoding/json"
	"os"
	"os/signal"
	"sync/atomic"
	"syscall"
	"time"

	m "gitee.com/sting/ksync/messages"

	_ "github.com/lib/pq"
)

var (
	postgresDB Database
	callback   *m.ConsumerCallbacks

	totalCT        int32
	preTotalCT     int32
	processingCT   int32
	throughput     int32
	gDynamicNumber int32 = 2000
	gStep          int32 = 100
)

func Sync(brokersList []string, topic string, consumerGroup string, dburl string, fromBegin bool) {
	// sarama.Logger = log.New(os.Stderr, "", log.LstdFlags)

	if dburl != "" {
		postgresDB = Database{
			driver: "postgres",
			url:    dburl,
		}
		postgresDB.open()
		callback = DBProcesser(func(record *m.DBEventLog) {
			if err := postgresDB.saveEvent(record); err != nil {
				Log.Errorf("error insert to the database: %v\n%+v", err, record)
			}
		})
	} else {
		callback = DBProcesser(func(record *m.DBEventLog) {
			Log.Debugf("verbose message: %+v", record)
		})
	}

	consumer := m.NewConsumer(*callback, brokersList, consumerGroup, []string{topic}, fromBegin)
	consumer.Consume()

	signals := make(chan os.Signal, 1)
	signal.Notify(signals, os.Interrupt, os.Kill, syscall.SIGHUP)

	go func() {
		for range time.Tick(1 * time.Second) {
			if totalCT > preTotalCT {
				throughput = (totalCT - preTotalCT - processingCT) //throughput < 0 大量积压
				preTotalCT = totalCT
				Log.Debugf("stats:[pre]%v\t[total]%v\t[processing]%v,\t[gDynamicNumber] %v,\t[throughput] %v", preTotalCT, totalCT, processingCT, gDynamicNumber, throughput)
				if processingCT > 0 {
					if throughput <= 0 || (processingCT > throughput && processingCT/throughput >= 2) {
						step := (processingCT - throughput) / 10 //十分之一缩减
						if gDynamicNumber > step && gDynamicNumber-step > 200 {
							gDynamicNumber = gDynamicNumber - step
						} else {
							step = gDynamicNumber - 200
							gDynamicNumber = 200
						}

						Log.Debugf("speed down,decrease %v,\t[processing]%v,\t[gDynamicNumber] %v,\t[throughput] %v", step, processingCT, gDynamicNumber, throughput)
					}

					if throughput > processingCT && throughput/processingCT >= 2 {
						gDynamicNumber = gDynamicNumber + gStep
						Log.Debugf("speed up,  add %v go-routine,\t[processing]%v,\t[gDynamicNumber] %v,\t[throughput] %v", gStep, processingCT, gDynamicNumber, throughput)
					}
				}

			}
		}
	}()

endless:
	for {
		select {
		case <-signals:
			consumer.Close()
			Log.Infof(`
				Interrupt is detected
				wait processing message...  %v
				remain message... %v
				`, processingCT, len(m.BufferedMessages))

			doneCh := make(chan struct{}) //结束程序
			go func() {
				for range time.Tick(1 * time.Second) {
					Log.Infof("messages:[gDynamicNumber] %v \t [processing] %v \t [remain] %v", gDynamicNumber, processingCT, len(m.BufferedMessages))
					if processingCT <= 0 && len(m.BufferedMessages) == 0 {
						doneCh <- struct{}{}
					}
				}
			}()
			<-doneCh
			break endless
		}
	}
}

func DBProcesser(fn func(*m.DBEventLog)) *m.ConsumerCallbacks {

	consumerCallbacks := &m.ConsumerCallbacks{
		OnDataReceived: func(msg []byte) {
			var record m.DBEventLog
			err := json.Unmarshal(msg, &record)
			if err == nil {
				//控制g数量 gDynamicNumber
				if processingCT >= gDynamicNumber {
					Log.Debugf("pause launch goroutine,[control line] %v ", gDynamicNumber)
					ticker := time.NewTicker(time.Millisecond * 500)
					doneChan := make(chan bool)
					go func() {
						for _ = range ticker.C {
							if processingCT < gDynamicNumber {
								doneChan <- true
							}
						}
					}()
					<-doneChan
					Log.Debugf("resume start goroutine")
				}
				go func() {
					atomic.AddInt32(&processingCT, 1)
					atomic.AddInt32(&totalCT, 1)
					fn(&record)
					atomic.AddInt32(&processingCT, -1)
				}()
			} else {
				Log.Errorf("error unmarshal: %s [%s]", err, string(msg))
			}
		},
		OnError: func(err error) {
			Log.Errorf("onConsumerError: %v", err)
		},
	}
	return consumerCallbacks
}
